package cn.gwm.flink.streaming.function.sink;
import cn.gwm.flink.streaming.beans.trip.CarTripOutput;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.IdUtil;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.ds.DSFactory;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.LinkedList;


public class SinkHive extends RichSinkFunction<LinkedList<CarTripOutput>> {

    @Override
    public void invoke(LinkedList<CarTripOutput> carTripOutputs, Context context) throws Exception {
        super.invoke(carTripOutputs, context);
        for (CarTripOutput carTripOutput : carTripOutputs
        ) {
            carTripOutput = CarTripOutput.convertDecimalFormat(carTripOutput);
            carTripOutput = CarTripOutput.calculationAverageSpeed(carTripOutput);
            Entity entity = Entity.parse(carTripOutput, true, true).setTableName(carTripOutput.getTriptype());
            if (entity.containsKey("triptype")) {
                entity.remove("triptype");
            }
            try {
                DataSource group_db2 = DSFactory.get("group_db2");
                Db.use(group_db2).insert(entity);
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }
    }
}